feat(sync): bounded tail-first thread sync (v2) with media externalization and transport negotiation#3849
feat(sync): bounded tail-first thread sync (v2) with media externalization and transport negotiation#3849snipemanmike wants to merge 1 commit into
Conversation
|
Important Review skippedAuto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
size:XXL
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
| updatedAt: head.updatedAt, | ||
| archivedAt: head.archivedAt, | ||
| deletedAt: head.deletedAt, | ||
| messages: window.messages, |
There was a problem hiding this comment.
🟠 High state/windowedThread.ts:84
fromWindowSnapshot copies window.messages directly into state without stripping the ["Older content omitted from the synced window."] marker from truncated entries. When a streaming assistant message arrives via thread.message-sent, upsertMessage appends the new chunk onto the truncated entry.text, producing corrupted content with the omission marker embedded in the middle and permanently missing bytes. Consider replacing the marker with an empty string (or a sentinel the reducer recognizes) for messages that are still streaming so appended chunks concatenate onto the real prefix rather than the marker.
| messages: window.messages, | |
| messages: window.messages.map((message) => | |
| message.text === '["Older content omitted from the synced window."]' && message.streaming | |
| ? { ...message, text: '' } | |
| : message | |
| ), |
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/windowedThread.ts around line 84:
`fromWindowSnapshot` copies `window.messages` directly into state without stripping the `["Older content omitted from the synced window."]` marker from truncated entries. When a streaming assistant message arrives via `thread.message-sent`, `upsertMessage` appends the new chunk onto the truncated `entry.text`, producing corrupted content with the omission marker embedded in the middle and permanently missing bytes. Consider replacing the marker with an empty string (or a sentinel the reducer recognizes) for messages that are still streaming so appended chunks concatenate onto the real prefix rather than the marker.
There was a problem hiding this comment.
🟡 Medium
t3code/apps/web/src/connection/storage.ts
Line 545 in 38e4de8
Bumping StoredThreadSnapshot.schemaVersion from 2 to 3 makes every existing web thread-cache entry written as version 2 fail to decode. The loadThread path catches the decode error and returns a cold cache, but it never removes the stale record, so every subsequent open of that thread re-decodes the same incompatible entry and hits the error path again until the thread is eventually resaved. Consider deleting the stored value when decoding fails so the incompatible entry is evicted on first access.
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/web/src/connection/storage.ts around line 545:
Bumping `StoredThreadSnapshot.schemaVersion` from `2` to `3` makes every existing web thread-cache entry written as version `2` fail to decode. The `loadThread` path catches the decode error and returns a cold cache, but it never removes the stale record, so every subsequent open of that thread re-decodes the same incompatible entry and hits the error path again until the thread is eventually resaved. Consider deleting the stored value when decoding fails so the incompatible entry is evicted on first access.
| const nodeGzipCodec: CompressionCodec = { | ||
| compressSync: (b) => zlib.gzipSync(b, { level: 4 }), | ||
| decompressSync: (b) => zlib.gunzipSync(b), | ||
| threshold: 1024, | ||
| }; |
There was a problem hiding this comment.
🟠 High src/ws.ts:2151
nodeGzipCodec.decompressSync calls zlib.gunzipSync without maxOutputLength, so any authenticated client on /ws-compressed can send a small gzip-compressed binary frame that decompresses to a buffer large enough to block the event loop or crash the process via OOM before Node's default buffer.kMaxLength limit kicks in. Consider passing maxOutputLength to gunzipSync with a sane cap sized to the largest legitimate message.
| const nodeGzipCodec: CompressionCodec = { | |
| compressSync: (b) => zlib.gzipSync(b, { level: 4 }), | |
| decompressSync: (b) => zlib.gunzipSync(b), | |
| threshold: 1024, | |
| }; | |
| const nodeGzipCodec: CompressionCodec = { | |
| compressSync: (b) => zlib.gzipSync(b, { level: 4 }), | |
| decompressSync: (b) => zlib.gunzipSync(b, { maxOutputLength: 1 << 20 }), | |
| threshold: 1024, | |
| }; |
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/ws.ts around lines 2151-2155:
`nodeGzipCodec.decompressSync` calls `zlib.gunzipSync` without `maxOutputLength`, so any authenticated client on `/ws-compressed` can send a small gzip-compressed binary frame that decompresses to a buffer large enough to block the event loop or crash the process via OOM before Node's default `buffer.kMaxLength` limit kicks in. Consider passing `maxOutputLength` to `gunzipSync` with a sane cap sized to the largest legitimate message.
| case "project.created": | ||
| case "project.meta-updated": | ||
| case "project.deleted": | ||
| case "thread.created": |
There was a problem hiding this comment.
🟡 Medium state/windowedThread.ts:417
The thread.turn-diff-completed event falls through to { kind: "unchanged" } in applyWindowedThreadEvent, so in v2 mode every live checkpoint update is silently dropped. The caller treats unchanged as a no-op and advances the sequence, so the client never records the new checkpoint or the latest-turn settlement that the legacy reducer performs for this event. Consider handling thread.turn-diff-completed to update checkpoints and latestTurn (or returning { kind: "resync" } if the windowed state cannot reconstruct the checkpoint incrementally).
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/windowedThread.ts around line 417:
The `thread.turn-diff-completed` event falls through to `{ kind: "unchanged" }` in `applyWindowedThreadEvent`, so in v2 mode every live checkpoint update is silently dropped. The caller treats `unchanged` as a no-op and advances the sequence, so the client never records the new checkpoint or the latest-turn settlement that the legacy reducer performs for this event. Consider handling `thread.turn-diff-completed` to update `checkpoints` and `latestTurn` (or returning `{ kind: "resync" }` if the windowed state cannot reconstruct the checkpoint incrementally).
| SELECT activity_id FROM latest_user_input WHERE kind = 'user-input.requested' | ||
| ) | ||
| ) | ||
| ORDER BY activity.created_at ASC, activity.activity_id ASC |
There was a problem hiding this comment.
🟡 Medium Layers/ThreadSyncQuery.ts:339
getTail serializes every unresolved approval and pending user-input activity for the thread into head.pendingRequests with no limit on the number returned. listPendingRequests has no LIMIT clause, so a thread with many outstanding requests produces an unbounded pendingRequests array in the snapshot — defeating the size cap this PR introduces and potentially recreating the large-payload problem it avoids for messages and activities. Consider adding a LIMIT to listPendingRequests (e.g. matching the activity tail limit) so the pending-requests portion of the snapshot is also bounded.
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration/Layers/ThreadSyncQuery.ts around line 339:
`getTail` serializes every unresolved approval and pending user-input activity for the thread into `head.pendingRequests` with no limit on the number returned. `listPendingRequests` has no `LIMIT` clause, so a thread with many outstanding requests produces an unbounded `pendingRequests` array in the snapshot — defeating the size cap this PR introduces and potentially recreating the large-payload problem it avoids for messages and activities. Consider adding a `LIMIT` to `listPendingRequests` (e.g. matching the activity tail limit) so the pending-requests portion of the snapshot is also bounded.
ApprovabilityVerdict: Needs human review 14 blocking correctness issues found. This PR introduces a new thread sync protocol (v2) with compression, media externalization, and pagination across server and all clients. Multiple high-severity issues have been identified including potential DoS vulnerability, data corruption in streaming messages, incorrect activity ordering, and broken asset resolution, alongside numerous medium-severity bugs affecting core sync behavior. You can customize Macroscope's approvability policy. Learn more. |
…ation and transport negotiation Opening a large thread still transfers the entire thread body — every message, activity, and inline base64 blob since the thread began. On cellular this never completes for media-heavy threads, and parsing the resulting payload can OOM-kill the mobile app outright (observed: a 268MB thread snapshot). pingdotgg#3719 moved the snapshot off the socket; this change bounds what is transferred at all. Measured on a physical Android device over Tailscale: - One media-heavy thread's snapshot was ~25MB on the wire; gzip only buys 1.6x on inline base64. With the bounded tail it renders in seconds over 5G. - Sequence-cursor reconnects: 21,528KB re-sent per reconnect before, 36-54KB after (~400x). What this adds, all capability-gated so old peers are unaffected: - subscribeThreadV2: a small thread head (metadata, latest turn, session, active plan, pending approval/input requests, counts) plus the last 32 messages / 128 compact activities under a 512KiB inline budget, emitted as <=256KiB chunks with keepalives between them, then live events after the snapshot watermark. Catch-up replays are epoch-checked and bounded to the watermark; a bounded (2048) live buffer signals an explicit resync instead of queueing unboundedly. - getThreadHistoryPage: keyset pagination for older messages and activities (scroll-up on mobile), invalidated across thread.reverted via a history epoch so pages never mix pre/post-revert history. - Activity payload externalization: base64 data-URLs and oversized payloads are stored as content-addressed attachments served by the existing /api/assets route; the wire carries references and mobile renders tap-to-load placeholders. (Message attachments already work this way — activity payloads were the unbounded route.) - Transport negotiation: the environment descriptor advertises rpcTransports and threadSyncVersions (with decode defaults, so new clients read old descriptors); /ws stays byte-for-byte plain JSON and an optional gzip transport lives at /ws-compressed behind an optional RpcCompressionCodec (defaults to null; only opted-in platforms provide one). Cached-token reuse treats the negotiation probe as best-effort so offline reconnects keep working. - Client: a distinct windowed thread state (never passed through APIs typed as a complete OrchestrationThread), atomic snapshot staging committed together with its cursor, resyncs that restart the subscription with a logged reason, and thread caches that store either a legacy detail snapshot or a v2 window (older records evict as cache misses). Array-by-copy methods are avoided in mobile-bound code: Hermes on current retail devices lacks toSorted and fails as a silent fiber defect. Tests: v2 contract decode/round-trip, transactional tail and keyset page queries with revert invalidation, wire round-trip -> staging integration, windowed merge/revert reducers, mobile feed windowing, and an opt-in real-data replay suite (T3_THREAD_SYNC_REALDATA_DB) that replays production-scale threads through the pipeline. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
snipemanmike
force-pushed
the
pr/thread-sync-v2
branch
from
38e4de8 to
7a9e6e0
Compare
|
Updated (force-pushed) with all review findings addressed — thanks Bugbot/CodeRabbit, several were genuinely good catches:
Three findings kept as-is with reasoning in the inline replies (pako gzip autodetect, watermark buffering, whole-payload asset references). |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 7a9e6e0. Configure here.
| }, | ||
| }, | ||
| hasOlderMessages: snapshot.head.counts.messages > built.messages.length, | ||
| hasOlderActivities: snapshot.head.counts.activities > built.activities.length, |
There was a problem hiding this comment.
Older messages stuck without cursor
Medium Severity
The subscribeThreadV2 RPC's snapshot-complete item can incorrectly report hasOlderMessages as true while before.message is null. This occurs when the initial snapshot budget omits all messages. Consequently, getThreadHistoryPage cannot paginate to load older messages.
Reviewed by Cursor Bugbot for commit 7a9e6e0. Configure here.
| messages: window.messages, | ||
| activities: mergeActivities(window.activities, head.pendingRequests), | ||
| proposedPlans: head.activeProposedPlan === null ? [] : [head.activeProposedPlan], | ||
| checkpoints: [], |
There was a problem hiding this comment.
🟡 Medium state/windowedThread.ts:87
fromWindowSnapshot hard-codes checkpoints: [], so every v2/windowed thread loses all checkpoint summaries the moment it is loaded from a v2 snapshot or restored from cache. Downstream consumers reading thread.checkpoints directly see an empty array for all windowed threads, so checkpoint-based features disappear instead of showing server state. The WindowedOrchestrationThread type has no checkpoints field, so the data is never carried through. If checkpoints are intentionally excluded from the windowed sync payload, consider documenting that rationale; otherwise thread the server-provided checkpoints through fromWindowSnapshot (and toWindowSnapshot) so they survive round-trips.
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/windowedThread.ts around line 87:
`fromWindowSnapshot` hard-codes `checkpoints: []`, so every v2/windowed thread loses all checkpoint summaries the moment it is loaded from a v2 snapshot or restored from cache. Downstream consumers reading `thread.checkpoints` directly see an empty array for all windowed threads, so checkpoint-based features disappear instead of showing server state. The `WindowedOrchestrationThread` type has no `checkpoints` field, so the data is never carried through. If checkpoints are intentionally excluded from the windowed sync payload, consider documenting that rationale; otherwise thread the server-provided checkpoints through `fromWindowSnapshot` (and `toWindowSnapshot`) so they survive round-trips.
| const listTailActivities = SqlSchema.findAll({ | ||
| Request: ThreadLookup, | ||
| Result: ProjectionActivityDbRow, | ||
| execute: ({ threadId }) => sql` | ||
| SELECT activity_id AS "activityId", thread_id AS "threadId", turn_id AS "turnId", tone, | ||
| kind, summary, payload_json AS payload, sequence, created_at AS "createdAt" | ||
| FROM projection_thread_activities WHERE thread_id = ${threadId} | ||
| ORDER BY created_at DESC, activity_id DESC LIMIT ${THREAD_SYNC_TAIL_ACTIVITY_LIMIT} | ||
| `, | ||
| }); |
There was a problem hiding this comment.
🟠 High Layers/ThreadSyncQuery.ts:297
listTailActivities and listActivityPage order activities by created_at DESC, activity_id DESC instead of sequence DESC, activity_id DESC. When activities arrive with out-of-order createdAt values, these queries return them in timestamp order rather than event-sequence order, so clients receive a thread history that does not match the actual event sequence and paginate around the wrong cursor. Consider ordering by sequence DESC, activity_id DESC to match the existing projection snapshot behavior.
const listTailActivities = SqlSchema.findAll({
Request: ThreadLookup,
Result: ProjectionActivityDbRow,
execute: ({ threadId }) => sql`
SELECT activity_id AS "activityId", thread_id AS "threadId", turn_id AS "turnId", tone,
kind, summary, payload_json AS payload, sequence, created_at AS "createdAt"
FROM projection_thread_activities WHERE thread_id = ${threadId}
- ORDER BY created_at DESC, activity_id DESC LIMIT ${THREAD_SYNC_TAIL_ACTIVITY_LIMIT}
+ ORDER BY sequence DESC, activity_id DESC LIMIT ${THREAD_SYNC_TAIL_ACTIVITY_LIMIT}
`,
});🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration/Layers/ThreadSyncQuery.ts around lines 297-306:
`listTailActivities` and `listActivityPage` order activities by `created_at DESC, activity_id DESC` instead of `sequence DESC, activity_id DESC`. When activities arrive with out-of-order `createdAt` values, these queries return them in timestamp order rather than event-sequence order, so clients receive a thread history that does not match the actual event sequence and paginate around the wrong cursor. Consider ordering by `sequence DESC, activity_id DESC` to match the existing projection snapshot behavior.
| case "resync-required": | ||
| yield* requestV2Resync(`server-resync-required (${item.reason})`); | ||
| return; | ||
| case "keepalive": |
There was a problem hiding this comment.
🟡 Medium state/threads.ts:514
The keepalive handler in applyV2Item sets status to "live" unconditionally (except when already "deleted"), so a keepalive received before the first snapshot-complete transitions the thread to live while data is still Option.none() (or holds only stale cached data). subscribeThreadV2 is allowed to send keepalives between snapshot chunks, so on a cold open the UI sees live status before any v2 snapshot has been committed. Consider guarding the keepalive case to only set "live" when the thread already has data (e.g. Option.isSome(current.data)), leaving the status unchanged otherwise.
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/threads.ts around line 514:
The `keepalive` handler in `applyV2Item` sets `status` to `"live"` unconditionally (except when already `"deleted"`), so a keepalive received before the first `snapshot-complete` transitions the thread to `live` while `data` is still `Option.none()` (or holds only stale cached data). `subscribeThreadV2` is allowed to send keepalives between snapshot chunks, so on a cold open the UI sees `live` status before any v2 snapshot has been committed. Consider guarding the `keepalive` case to only set `"live"` when the thread already has data (e.g. `Option.isSome(current.data)`), leaving the status unchanged otherwise.
| import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; | ||
| import * as RelayClient from "@t3tools/shared/relayClient"; | ||
| const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); | ||
| const isOrchestrationGetSnapshotError = Schema.is(OrchestrationGetSnapshotError); |
There was a problem hiding this comment.
🟡 Medium src/ws.ts:133
Schema.is(OrchestrationGetSnapshotError) at line 133 returns false for OrchestrationGetSnapshotError instances that getThreadHistoryPage constructs with a plain object cause ({ requestedHistoryEpoch, currentHistoryEpoch }), because the schema declares cause as Schema.Defect() which does not validate plain objects. As a result, the Effect.mapError path wraps the original error in a generic OrchestrationGetSnapshotError, discarding the specific "history changed; reload the tail window" message and epoch details that clients need to trigger a targeted resync. Consider using a looser predicate (e.g., checking the _tag discriminant) or adjusting the schema so object-shaped causes pass validation.
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/ws.ts around line 133:
`Schema.is(OrchestrationGetSnapshotError)` at line 133 returns `false` for `OrchestrationGetSnapshotError` instances that `getThreadHistoryPage` constructs with a plain object `cause` (`{ requestedHistoryEpoch, currentHistoryEpoch }`), because the schema declares `cause` as `Schema.Defect()` which does not validate plain objects. As a result, the `Effect.mapError` path wraps the original error in a generic `OrchestrationGetSnapshotError`, discarding the specific "history changed; reload the tail window" message and epoch details that clients need to trigger a targeted resync. Consider using a looser predicate (e.g., checking the `_tag` discriminant) or adjusting the schema so object-shaped causes pass validation.
| ? thread.latestTurn.assistantMessageId | ||
| : null, | ||
| } | ||
| : thread.latestTurn?.state === "running" |
There was a problem hiding this comment.
🟡 Medium state/windowedThread.ts:385
applyWindowedThreadEvent settles a running latestTurn to "completed" for every non-"error" thread.session-set event. When the session leaves running with status "interrupted" or "stopped", the turn is incorrectly reported as completed instead of interrupted. Consider mapping those statuses to "interrupted", matching the legacy reducer's behavior.
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/windowedThread.ts around line 385:
`applyWindowedThreadEvent` settles a running `latestTurn` to `"completed"` for every non-`"error"` `thread.session-set` event. When the session leaves running with status `"interrupted"` or `"stopped"`, the turn is incorrectly reported as completed instead of interrupted. Consider mapping those statuses to `"interrupted"`, matching the legacy reducer's behavior.
| return !forceSnapshot && | ||
| seq > 0 && | ||
| Option.isSome(current.data) && | ||
| isWindowedThread(current.data.value) | ||
| ? { | ||
| threadId, | ||
| sinceSequence: seq, | ||
| historyEpoch: current.data.value.historyEpoch, | ||
| } | ||
| : { threadId }; |
There was a problem hiding this comment.
🟡 Medium state/threads.ts:602
resolveV2SubscribeInput requires Option.isSome(current.data) before sending sinceSequence, so once the thread is deleted and data is Option.none(), every reconnect sends only { threadId } without sinceSequence. The server treats a missing thread without a cursor as not found instead of replaying the terminal delete, so the client enters an error/retry loop on each reconnect instead of staying deleted. Consider allowing sinceSequence to be sent when the status is "deleted" (e.g. checking seq > 0 && (Option.isNone(current.data) || isWindowedThread(current.data.value))), so the server replays the delete event instead of returning not found.
| return !forceSnapshot && | |
| seq > 0 && | |
| Option.isSome(current.data) && | |
| isWindowedThread(current.data.value) | |
| ? { | |
| threadId, | |
| sinceSequence: seq, | |
| historyEpoch: current.data.value.historyEpoch, | |
| } | |
| : { threadId }; | |
| return !forceSnapshot && | |
| seq > 0 && | |
| (Option.isNone(current.data) || isWindowedThread(current.data.value)) | |
| ? { | |
| threadId, | |
| sinceSequence: seq, | |
| historyEpoch: Option.isSome(current.data) && isWindowedThread(current.data.value) | |
| ? current.data.value.historyEpoch | |
| : undefined, | |
| } | |
| : { threadId }; |
🤖 Copy this AI Prompt to have your agent fix this:
In file @packages/client-runtime/src/state/threads.ts around lines 602-611:
`resolveV2SubscribeInput` requires `Option.isSome(current.data)` before sending `sinceSequence`, so once the thread is deleted and `data` is `Option.none()`, every reconnect sends only `{ threadId }` without `sinceSequence`. The server treats a missing thread without a cursor as `not found` instead of replaying the terminal delete, so the client enters an error/retry loop on each reconnect instead of staying deleted. Consider allowing `sinceSequence` to be sent when the status is `"deleted"` (e.g. checking `seq > 0 && (Option.isNone(current.data) || isWindowedThread(current.data.value))`), so the server replays the delete event instead of returning `not found`.
There was a problem hiding this comment.
🟡 Medium
The payloadAsset filter in ThreadWorkLog silently drops media-bearing rows: activities with status === "neutral" and toolLike === true are removed before rendering, so any attachment on an in-progress or informational tool activity is never shown and the user has no way to open it. Consider preserving rows that carry a payloadAsset regardless of their status, or excluding only toolLike && status === "neutral" rows when they have no asset.
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/mobile/src/features/threads/ThreadFeed.tsx around line 139:
The `payloadAsset` filter in `ThreadWorkLog` silently drops media-bearing rows: activities with `status === "neutral"` and `toolLike === true` are removed before rendering, so any attachment on an in-progress or informational tool activity is never shown and the user has no way to open it. Consider preserving rows that carry a `payloadAsset` regardless of their status, or excluding only `toolLike && status === "neutral"` rows when they have no asset.
| // on the wire, impossible to corrupt. | ||
| return { | ||
| bytes: Buffer.from(data, "utf8"), | ||
| mediaType: "text/plain; charset=utf-8", |
There was a problem hiding this comment.
🟠 High orchestration/ThreadSyncWire.ts:109
externalizeActivityPayload externalizes large plain-text payloads as .txt files, but the attachment resolver only recognizes image extensions and .bin. Those asset references always 404 when the client requests them from /api/assets, silently making large text payloads unloadable. Consider using .bin for the plain-text fallback in externalizableData, or extend the attachment resolver to recognize .txt.
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration/ThreadSyncWire.ts around line 109:
`externalizeActivityPayload` externalizes large plain-text payloads as `.txt` files, but the attachment resolver only recognizes image extensions and `.bin`. Those asset references always 404 when the client requests them from `/api/assets`, silently making large text payloads unloadable. Consider using `.bin` for the plain-text fallback in `externalizableData`, or extend the attachment resolver to recognize `.txt`.
1 participant
Problem
Opening a large thread still transfers the entire thread body — every message, activity, and inline base64 blob since the thread began. #3719 moved the snapshot off the socket (great!), but the payload itself is still unbounded. Measured on a physical Android device over Tailscale:
Failed to allocate a 268501000 byte allocation).afterSequence, which this builds on.With this change the same media-heavy thread renders in seconds over 5G, and live send/receive works normally on threads of any size.
What this adds
Everything is capability-gated via the environment descriptor, so old/new client-server pairs interoperate; the existing flows are untouched when v2 isn't negotiated.
subscribeThreadV2— a small thread head (metadata, latest turn, session, active proposed plan, pending approval/input requests, counts) plus the last 32 messages / 128 compact activities under a 512 KiB inline budget, emitted as ≤256 KiB chunks with keepalives between them, then live events strictly after the snapshot watermark. Catch-ups are epoch-checked and watermark-bounded; a bounded (2,048) live buffer emits an explicitresync-requiredinstead of queueing unboundedly behind a slow cellular consumer.getThreadHistoryPage— keyset pagination for older messages/activities (scroll-up on mobile), guarded by a history epoch so pages can never mix pre/post-thread.revertedhistory./api/assetsroute; the sync stream carries references and mobile renders tap-to-load placeholders. (Message attachments already work this way — activity payloads were the unbounded route.)rpcTransportsandthreadSyncVersions(with decode defaults so new clients read old descriptors)./wsstays byte-for-byte plain JSON; an optional gzip transport lives at/ws-compressedbehind anRpcCompressionCodecreference that defaults tonull(only opted-in platforms provide one). Cached-token reuse treats the negotiation probe as best-effort so offline reconnects keep working.WindowedOrchestrationThread(never passed through APIs typed as a completeOrchestrationThread), atomic snapshot staging committed together with its cursor, resyncs that restart the subscription with a logged reason, and thread caches that store either a legacy detail snapshot or a v2 window (older records evict as cache misses).Mobile-specific finding worth knowing
Hermes on current retail Android devices lacks
Array.prototype.toSorted— it fails as a silent fiber defect rather than a visible error. This PR avoids array-by-copy methods in mobile-bound code; there's an existingtoReversedinapps/mobile/src/features/keyboard/hardwareKeyboardCommands.tsthat may be affected too.Tests
Contract decode/wire round-trips, transactional tail + keyset page queries with revert invalidation, a wire-round-trip → staging integration test, windowed merge/revert reducers, mobile feed windowing, and an opt-in production-scale replay suite (
T3_THREAD_SYNC_REALDATA_DB=path/to/state.sqlite) that pushes real thread data through the full pipeline.Happy to split this into smaller PRs (externalization / windowed sync / negotiation) if you'd prefer — and to rework the v2 tail as an extension of the #3719 HTTP snapshot endpoint instead of a WS RPC if that fits your direction better.
🤖 Generated with Claude Code
Note
High Risk
Large cross-cutting change to orchestration sync, WebSocket RPC transport, and persistence/cache formats; incorrect staging, epoch, or catch-up logic could cause silent hangs, stale history, or broken reconnects on production-scale threads.
Overview
This PR replaces full-thread downloads with a bounded tail-first sync (v2) when the environment descriptor advertises
threadSyncVersions: [1, 2]. The server exposessubscribeThreadV2(chunked snapshot start/chunk/complete, optional incremental catch-up, then live events withresync-requiredon overflow/revert) andgetThreadHistoryPage(keyset paging guarded by a revert history epoch).ThreadSyncQueryreads bounded tails and pages from SQLite;ThreadSyncWiretrims/chunks snapshots and externalizes oversized activity payloads to attachment references.Transport negotiation adds descriptor
rpcTransports(/wsplain JSON vs/ws-compressedgzip-json),RpcCompressionCodec(pako on mobile/web, zlib on server), and client logic to pick transport/sync version and rewrite the WebSocket path. Durable subscriptions can pass freshsinceSequenceon reconnect for delta catch-up.Clients persist thread cache v3: legacy detail snapshots or
WindowedOrchestrationThread, with stale v2 records evicted on load. Mobile gains scroll-uploadOlder, tap-to-load message/work-log media with timeouts, and feed logic so v2 messages and activities paginate independently.Risk note: thread snapshot cache schema bumps to v3; older cached thread bodies are dropped and force a resync.
Reviewed by Cursor Bugbot for commit 7a9e6e0. Bugbot is set up for automated code reviews on this repo. Configure here.
Note
Add bounded tail-first v2 thread sync with gzip transport negotiation and media externalization
subscribeThreadV2) that streams a bounded tail window (32 messages / 128 activities) to clients, withloadOlder/hasOlderpagination via a newgetThreadHistoryPageRPC./ws(JSON) and/ws-compressed(gzip-JSON) endpoints; clients select the best transport based on codec availability usingpako(web/mobile) ornode:zlib(server).ThreadSyncQueryLiveon the server to query projection tables for thread tails and paginated history, backed by a new per-threadreadThreadFromSequencestream on the event store.windowedThread.tson the client withWindowedOrchestrationThreadState, converters (fromWindowSnapshot/toWindowSnapshot), andmergeWindowHistoryPagefor history merging.Macroscope summarized 7a9e6e0.